package main

//
//import (
//	"bufio"
//	"encoding/binary"
//	"fmt"
//	"io"
//	"net"
//	"rpc/msg"
//	"strconv"
//	"sync"
//	"time"
//)
//
//var (
//	pool = &sync.Pool{
//		New: func() interface{} {
//			return make([]byte, 0, 10240)
//		}}
//
//	client8bPool = &sync.Pool{
//		New: func() interface{} {
//			return make([]byte, 8)
//		}}
//)
//
//func main() {
//
//	var wg sync.WaitGroup
//
//	client := NewClient(1, "127.0.0.1:8080")
//
//	now := time.Now()
//	for i := 0; i < 100000; i++ {
//		wg.Add(1)
//		go func() {
//			_, _ = client.Send([]byte("this is 2015 year" + strconv.Itoa(i)))
//			//fmt.Println(string(v))
//			wg.Done()
//		}()
//	}
//
//	wg.Wait()
//	fmt.Println("elapse: ", time.Since(now))
//
//}
//
//const (
//	BatchSize  = 1 * 1024 * 1024
//	BatchSize2 = 10000
//)
//
//type Client struct {
//	connection []*Conn
//	ack        *sync.Map
//	writeLock  chan struct{}
//	sendMerge  chan *msg.Msg
//}
//
//type Conn struct {
//	con            net.Conn
//	lock           *sync.Mutex
//	state          int
//	waitSendPacket chan *msg.Packet
//}
//
//func NewClient(connPool int, addr string) *Client {
//	cli := Client{
//		connection: make([]*Conn, 0),
//		ack:        new(sync.Map),
//		writeLock:  make(chan struct{}, 1),
//		sendMerge:  make(chan *msg.Msg, 0),
//	}
//
//	for i := 0; i < connPool; i++ {
//		conn, err := net.Dial("tcp4", addr)
//		if err != nil {
//			panic(err)
//		}
//		cli.connection = append(cli.connection, &Conn{
//			con:            conn,
//			lock:           new(sync.Mutex),
//			state:          0,
//			waitSendPacket: make(chan *msg.Packet, 102400),
//		})
//	}
//
//	cli.doRead()
//	cli.doSend2()
//
//	return &cli
//}
//
//func (client *Client) Send(data []byte) ([]byte, error) {
//	message := msg.NewMsg(data)
//	resultChan := make(chan *msg.Msg)
//	client.ack.Store(message.SeqNo, resultChan)
//	defer func() {
//		client.ack.Delete(message.SeqNo)
//	}()
//	select {
//	case client.sendMerge <- message:
//		// 请求被合并了，等待结果
//		rep := <-resultChan
//		return rep.Data, nil
//	case client.writeLock <- struct{}{}:
//
//	}
//	client.SendWithLock(message)
//
//	rep := <-resultChan
//	return rep.Data, nil
//}
//
//func (client *Client) SendWithLock(request *msg.Msg) {
//	var batch = []*msg.Msg{request}
//	mergeLimit := uint64(BatchSize)
//merge:
//	for mergeLimit > 0 {
//		select {
//		case msg := <-client.sendMerge:
//			if msg.Data != nil {
//				if msg.MsgTotalLength() > mergeLimit {
//					fmt.Println(len(batch))
//					break merge
//				}
//				batch = append(batch, msg)
//				mergeLimit = mergeLimit - msg.MsgTotalLength()
//			}
//		default:
//			break merge
//		}
//	}
//	<-client.writeLock
//	fmt.Println(len(batch))
//	client.asyncSendPacket(batch)
//}
//
//func (client *Client) asyncSendPacket(message []*msg.Msg) {
//	i := time.Now().UnixMicro() % int64(len(client.connection))
//	waitSendPacket := client.connection[i].waitSendPacket
//	packet := make([]byte, 0, 10240)
//	for _, ms := range message {
//		packet = append(packet, ms.Encode()...)
//	}
//	waitSendPacket <- msg.NewPacket(&packet)
//}
//
//func (client *Client) doRead() {
//	for _, con := range client.connection {
//		go func(conn *Conn) {
//			reader := bufio.NewReader(conn.con)
//			for {
//				header := client8bPool.Get().([]byte)
//				_, err := io.ReadFull(reader, header)
//				if err != nil {
//					if err == io.EOF {
//						return
//					}
//					return
//				}
//				l := binary.BigEndian.Uint64(header[:8])
//
//				client8bPool.Put(header)
//
//				data := make([]byte, l)
//				_, err = io.ReadFull(reader, data)
//				if err != nil {
//					if err == io.EOF {
//						return
//					}
//					return
//				}
//				p := &msg.Packet{
//					Header: l,
//					Batch:  &data,
//				}
//				go client.handlePacket(p)
//			}
//		}(con)
//	}
//}
//
//func (client *Client) handlePacket(packet *msg.Packet) {
//	batch := *packet.Batch
//	i := 0
//	for i < len(batch) {
//		l := binary.BigEndian.Uint64(batch[i : 8+i])
//		seqNo := binary.BigEndian.Uint64(batch[8+i : 16+i])
//		endIdx := 16 + i + int(l) - 8
//		req := &msg.Msg{
//			Length: l,
//			SeqNo:  seqNo,
//			Data:   batch[16+i : endIdx],
//		}
//		i = endIdx
//		client.asyncHandelResponse(req)
//	}
//}
//
//func (client *Client) asyncHandelResponse(ms *msg.Msg) {
//	value, _ := client.ack.Load(ms.SeqNo)
//	repChan, ok := value.(chan *msg.Msg)
//	if !ok {
//		fmt.Println("+++++", ms.SeqNo, *ms)
//	}
//	repChan <- ms
//}
//
//func (client *Client) doSend2() {
//	for _, con := range client.connection {
//		go func(conn *Conn) {
//			for {
//				select {
//				case p := <-conn.waitSendPacket:
//					conn.con.Write(p.Encode())
//				}
//			}
//		}(con)
//	}
//}
